一、說明

老樣子，又到了喜聞樂見的程式整理環節，我們今天會完成下列事項

• 資料拆分：以1:1比例分成訓練集與驗證集。
• 報表顯示：包括年化報酬率、最大跌幅等日後分析與視覺化時需要的資料。
• 參數優化

訓練器

主要作用為策略回測，策略參數優化，回測完畢後回傳該股票的獲利與風險報表。

def bt_trainer(stock_index, test_strategy, opt=False, opt_params=None):
    print(stock_index)

    train_df = load_stock(stock_index, start_year=2012, end_year=2015)
    if len(train_df) == 0:
        return pd.DataFrame()

    valid_df = load_stock(stock_index, start_year=2016, end_year=2019)
    if len(valid_df) == 0:
        return pd.DataFrame()

    train_holding_cash = (
        1000
        * train_df.groupby(pd.DatetimeIndex(train_df.index).to_period("M"))
        .nth(0)["Close"]
        .sum()
    )

    valid_holding_cash = (
        1000
        * valid_df.groupby(pd.DatetimeIndex(valid_df.index).to_period("M"))
        .nth(0)["Close"]
        .sum()
    )

    train_bt = Backtest(
        train_df,
        test_strategy,
        cash=train_holding_cash,
        commission=0.004,
        # exclusive_orders=True,
        trade_on_close=True,
    )

    valid_bt = Backtest(
        valid_df,
        test_strategy,
        cash=valid_holding_cash,
        commission=0.004,
        # exclusive_orders=True,
        trade_on_close=True,
    )

    if opt:
        train_result = train_bt.optimize(**opt_params)

        valid_result = valid_bt.run(**train_result["_strategy"].params)
    else:
        train_result = train_bt.run()
        valid_result = valid_bt.run()

    result = train_result.copy()
    result["stock_id"] = stock_index
    result["Strategy"] = test_strategy.__name__
    result["Params"] = result["_strategy"].params
    result["Train Return (Ann.) [%]"] = train_result["Return (Ann.) [%]"]
    result["Train Max. Drawdown [%]"] = train_result["Max. Drawdown [%]"]
    result["Train Sharpe Ratio"] = train_result["Sharpe Ratio"]
    result["Train SQN"] = train_result["SQN"]
    result["Valid Return (Ann.) [%]"] = valid_result["Return (Ann.) [%]"]
    result["Valid Max. Drawdown [%]"] = valid_result["Max. Drawdown [%]"]
    result["Valid Sharpe Ratio"] = valid_result["Sharpe Ratio"]
    result["Valid SQN"] = valid_result["SQN"]
    # result["Backtest Train"] = train_bt
    # result["Backtest Valid"] = valid_bt

    opt_str = "opt" if opt else "std"
    plot_dir = f"./data/Strategy/{test_strategy.__name__}/{opt_str}"
    Path(plot_dir).mkdir(parents=True, exist_ok=True)
    valid_bt.plot(
        filename=plot_dir + f"/{stock_index}.html",
        open_browser=False,
    )

    df = result.to_frame().transpose()

    return df

報表整理

def report_generate(frames, test_strategy, opt):
    final_df = pd.concat(frames).reset_index(drop=True)

    final_df = final_df.loc[
        final_df["Duration"] == final_df["Duration"].max()
    ].reset_index(drop=True)

    stock_name = pd.read_csv(
        "data/stock_id.csv", dtype={"stock_id": str, "stock_name": str}
    )

    profit_df = final_df.merge(stock_name, on="stock_id")

    report_df = profit_df[
        [
            "stock_id",
            "stock_name",
            "Train Return (Ann.) [%]",
            "Train Max. Drawdown [%]",
            "Train Sharpe Ratio",
            "Train SQN",
            "Valid Return (Ann.) [%]",
            "Valid Max. Drawdown [%]",
            "Valid Sharpe Ratio",
            "Valid SQN",
        ]
    ]

    opt_str = "opt" if opt else "std"

    report_dir = f"./data/Strategy/{test_strategy.__name__}/{opt_str}"
    Path(report_dir).mkdir(parents=True, exist_ok=True)

    report_df.to_csv(
        report_dir + "/Report.csv",
        index=False,
        header=True,
    )

    report_df = pd.read_csv(report_dir + "/Report.csv", dtype={"stock_id": str})

    return report_df

• 完整程式碼

二、結語

回測的部分差不多結束了，明天會測試幾個常見的幾種策略。